package com.letv.download;

import java.sql.Connection;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.letv.bean.SummaryBean;
import com.letv.common.ConnectionManager;
import com.letv.common.Constants;
import com.letv.common.ProcessData;
import com.letv.conf.ConfigFactory;
import com.letv.util.SQLUtil;

public class DownloadServiceImpl extends DownloadMapperService {

	private final static Logger log = LoggerFactory.getLogger(DownloadServiceImpl.class);

    /**
     * DownloadUtils, common method for download data from cms
     */
    private DownloadUtils downloadUtils = new DownloadUtils();

	public SummaryBean down(String type) {
		SummaryBean bean = new SummaryBean();

		log.info("begin down " + type + "...");

		// get DBconnection
		ConnectionManager cm = null;
		try {
			cm = ConnectionManager.getInstance();
		} catch (Exception e) {
			log.error("connection get instance error!", e);
			bean.setErrorMsg("connection get instance error! " + e.getMessage());
			return bean;
		}
		Connection conn = cm.getConnection();
		Connection connToSrc = null;

		if (Constants.INDEX_LESO.equals(type)) {
			connToSrc = cm.getConnectionFromLeso();
		} else {
			connToSrc = cm.getConnectionFromMMS();
		}

		// table备份
		log.info("copy " + type + " table...");

		if (Constants.INDEX_LESO.equals(type)) {
			this.downloadUtils.copyTable(conn, Constants.LESO_TB, Constants.LESO_OLD_TB);
		} else if (Constants.INDEX_ALBUM.equals(type)) {
		    this.downloadUtils.copyTable(conn, Constants.ALBUM_TB, Constants.ALBUM_OLD_TB);
		} else if (Constants.INDEX_STAR.equals(type)) {
		    this.downloadUtils.copyTable(conn, Constants.STAR_TB, Constants.STAR_OLD_TB);
		} else if (Constants.INDEX_VIDEO.equals(type)) {
		    this.downloadUtils.copyTable(conn, Constants.VIDEO_TB, Constants.VIDEO_OLD_TB);
		} else {
			log.warn(type + " data is not exist!");
			return bean;
		}

		// get minId and maxId
		ArrayList<String> idRange = new ArrayList<String>();
		try {
			if (Constants.INDEX_LESO.equals(type)) {
				idRange = SQLUtil.getIdRangeNewLeso(connToSrc);
			} else {
				idRange = SQLUtil.getSourceDataIdRange(connToSrc, type);
			}
		} catch (Exception e) {
			log.error("get " + type + " id range error!", e);
			bean.setErrorMsg("get " + type + " id range error! " + e.getMessage());
			return bean;
		} finally {
		    
		    boolean r1 = SQLUtil.close(connToSrc);
		    boolean r2 = SQLUtil.close(conn);
		    if(!r1 || !r2) {
                bean.setErrorMsg(bean.getErrorMsg() + " [ close db connections error! ");
		    }
		}

		Long minId = Long.valueOf(idRange.get(0)) - 1;
		Long maxId = Long.valueOf(idRange.get(1));

		log.debug("MaxId : " + maxId + ", MinId : " + minId);

		// put data by thread
		ExecutorService threadPool = Executors.newFixedThreadPool(ConfigFactory.getInt("threadPool.basesize", 30));
		List<DownloadWorker> workList = new ArrayList<DownloadWorker>();
		List<Future<DownloadResult>> results = new ArrayList<Future<DownloadResult>>();

		int pageSize = ConfigFactory.getInt("thread.pageSize", 3000);
		Long totalPage = 0l;
		totalPage = (maxId - minId) / pageSize + 1;
		log.info("type="+type+",totalPage : " + totalPage);

		for (int i = 1; i <= totalPage; i++) {
			log.debug("create thread : from id=" + (minId + pageSize * (i - 1)) + " to id=" + (minId + pageSize * i));
			DownloadWorker worker = null;

			if (Constants.INDEX_LESO.equals(type)) {
				worker = new DownloadWorker(Constants.LESO_TB, "", type, minId + pageSize * (i - 1), minId + pageSize * i);
			} else if (Constants.INDEX_ALBUM.equals(type)) {
				worker = new DownloadWorker(Constants.ALBUM_TB, "", type, minId + pageSize * (i - 1), minId + pageSize * i);
			} else if (Constants.INDEX_STAR.equals(type)) {
				worker = new DownloadWorker(Constants.STAR_TB, "", type, minId + pageSize * (i - 1), minId + pageSize * i);
			} else if (Constants.INDEX_VIDEO.equals(type)) {
				worker = new DownloadWorker(Constants.VIDEO_TB, "", type, minId + pageSize * (i - 1), minId + pageSize * i);
			} 

			workList.add(worker);
		}
//		Long[] tempIdArray = {10319L};
//		for(Long tempId : tempIdArray) {
//		    DownloadWorker worker = new DownloadWorker(Constants.VIDEO_TB, "", type, tempId, tempId + 1L);
//		    workList.add(worker);
//		}

		// get result of threads
		int successCnt = 0;
		int failCnt = 0;

		try {
			results = threadPool.invokeAll(workList);

			for (Future<DownloadResult> future : results) {
				try {
					DownloadResult r = future.get();
					if (!future.isCancelled()) {
						successCnt += r.getRecordsCnt();
						log.info("resolve " + r.getRecordsCnt() + " 条数据, result=" + r.isFlag());
					} else {
						failCnt += r.getRecordsCnt();
						log.info("cancel " + r.getRecordsCnt() + " 条数据, result=" + r.isFlag());
					}
				} catch (ExecutionException e) {
					log.error("get thread result ExecutionException!", e);
					bean.setErrorMsg(bean.getErrorMsg() + " [get thread result ExecutionException!"
					        + e.getMessage());
					continue;
				} catch (Exception e) {
					log.error("get thread result exception!", e);
					bean.setErrorMsg(bean.getErrorMsg() + " [ get thread result exception!"
					        + e.getMessage());
					continue;
				}
			}
		} catch (InterruptedException e) {
			log.error("excute " + type + " thread error!", e);
			bean.setErrorMsg(bean.getErrorMsg() + " [excute " + type + " thread error!"
			        + e.getMessage());
		} finally {
			threadPool.shutdown();
			log.info("threadPool " + type + " shutdown!");
		}

		log.info("download " + type + " done!");

		bean.setSubTaskMysqlSuccessedCountNum(successCnt);
		bean.setSubTaskMysqlFailedCountNum(failCnt);

		return bean;
	}

	public boolean downDataById(String type, Long id) {
		log.info("downloadImpl downDataById begin...");
		ConnectionManager cm = null;
		Connection conn = null;
		Connection connToSrc = null;
		try {
			cm = ConnectionManager.getInstance();
			log.info("get cm=" + cm.toString());
			conn = cm.getConnection();
			log.info("get conn=" + conn.toString());
			if (Constants.INDEX_LESO.equals(type)) {
				connToSrc = cm.getConnectionFromLeso();
			} else {
				connToSrc = cm.getConnectionFromMMS();
				log.info("get connToSrc=" + connToSrc.toString());
			}
			Map<String, Object> m = null;
			ArrayList<Map<String, Object>> lists = null;
			if (Constants.INDEX_LESO.equals(type)) {
		        lists = SQLUtil.getLesoAlbumSourceData(connToSrc, "leso.SQL2", id, id+1);
		        if (lists.size() ==1) {
		        	m = lists.get(0);
		        	Map<String,Object> tmp  = SQLUtil.getSourceDataById(connToSrc, type, Long.valueOf(m.get("aid").toString()));
					if (tmp ==null ||tmp.size() ==0 ) {
						log.info("video table non't have data,album_id="+m.get("aid").toString()+",meby empty album.");
						//return false;
					} else {
					    if (tmp.get("vid")!=null && (!"".equals(tmp.get("vid").toString()))) {
					        m.putAll(tmp);
					        String longMessage=super.mapperVideoList(connToSrc, m);
					        log.info(longMessage);
                            List<Map<String, Object>> otherSourceDetailList = super.getOtherSourceDetailList(connToSrc, m);
                            if(null != otherSourceDetailList) {
                                lists.addAll(otherSourceDetailList);
                            }
					    }
					}
				}
			} else {
				m = SQLUtil.getSourceDataById(connToSrc, type, id);	
			}
			if (m == null) {
				log.warn("download type=" + type + ", id=" + id + " error!");
				return false;
			}
			List<String> recordList = new ArrayList<String>();
			if (Constants.INDEX_ALBUM.equals(type)) {
			    recordList.add(ProcessData.getAlbumStr(m));
			} else if (Constants.INDEX_STAR.equals(type)) {
			    recordList.add(ProcessData.getStarStr_new(m));
			} else if (Constants.INDEX_VIDEO.equals(type)) {
			    recordList.add(ProcessData.getVideoStr(m));
			} else if (Constants.INDEX_LESO.equals(type)) {
			    for(Map<String, Object> albumDetail : lists) {
			        recordList.add(ProcessData.getLesoStr(albumDetail));
			    }
			}
			for(int i = 0; i < recordList.size(); i++) {
			    String record = recordList.get(i);
			    log.info("download type=" + type + ",sort=" + i + ",id=" + id + " record=" + record.length());
			    try{
			        SQLUtil.insertSingleSourceData(conn, type, record);
			    }catch (SQLException e){
			        log.error("insert data record=" + record + "! id=" + id + " sql error!",e);
			    }
			}
			return true;
		} catch (SQLException e) {
			log.error("insert data type=" + type + ", id=" + id + " sql error!", e);
			return false;
		} catch (Exception e) {
			log.error("insert data type=" + type + ", id=" + id + " error!", e);
			return false;
		} finally {
            SQLUtil.close(connToSrc);
            SQLUtil.close(conn);
		}
	}

	private void calculateCount(Map<String, Object> map, String column, int divid) {
		Object dividend = map.get(column);
		if (dividend != null) {
			if (StringUtils.isNumeric(dividend.toString())) {
				Long value = Long.valueOf(dividend.toString()) / divid;
				map.put(column, value.toString());
			}
		}
	}

	public boolean delDataById(String type, Long id) {
		log.info("downloadImpl delDataById begin...");
		ConnectionManager cm = null;
		try {
			cm = ConnectionManager.getInstance();
		} catch (Exception e) {
			log.error("connection get instance error!", e);
			return false;
		}
		Connection conn = cm.getConnection();

		try {
			SQLUtil.deleteDataById(conn, type, id);
			return true;
		} catch (SQLException e) {
			log.error("delete type=" + type + ", id=" + id + " error!");
			return false;
		} finally {
		    SQLUtil.close(conn);
		}
	}

	public boolean updDataById(String type, Long id) {
		log.info("downloadImpl updDataById begin...");
		if (delDataById(type, id)) {
			return downDataById(type, id);
		} else {
			log.error("update type=" + type + ", id=" + id + " error!");
		}
		return false;
	}

    public SummaryBean downLesoExtraet(String type) {
		SummaryBean bean = new SummaryBean();

		if (!ConfigFactory.getString(type + ".pushTarget").contains("DB")) {
			log.info(type + " push target doesn't contain db, download " + type + " cancelled!");
			return bean;
		}

		log.info("begin down " + type + "...");

		// get DBconnection
		ConnectionManager cm = null;
		try {
			cm = ConnectionManager.getInstance();
		} catch (Exception e) {
			log.error("connection get instance error!", e);
			bean.setErrorMsg("connection get instance error! " + e.getMessage());
			return bean;
		}
		Connection conn = cm.getConnection();
		Connection connToSrc = null;

		if (Constants.INDEX_LESO.equals(type) || Constants.INDEX_LESO_EXTRAET.equals(type)) {
			connToSrc = cm.getConnectionFromLeso();
		} else {
			connToSrc = cm.getConnectionFromMMS();
		}

		// table备份
		log.info("copy " + type + " table...");

		if (Constants.INDEX_LESO.equals(type)) {
		    this.downloadUtils.copyTable(conn, Constants.LESO_TB, Constants.LESO_OLD_TB);
		} else if (Constants.INDEX_ALBUM.equals(type)) {
		    this.downloadUtils.copyTable(conn, Constants.ALBUM_TB, Constants.ALBUM_OLD_TB);
		} else if (Constants.INDEX_STAR.equals(type)) {
		    this.downloadUtils.copyTable(conn, Constants.STAR_TB, Constants.STAR_OLD_TB);
		} else if (Constants.INDEX_VIDEO.equals(type)) {
		    this.downloadUtils.copyTable(conn, Constants.VIDEO_TB, Constants.VIDEO_OLD_TB);
		} else {
			log.warn(type + " data is not exist!");
			SQLUtil.close(conn, connToSrc);
			return bean;
		}

		// get minId and maxId
		ArrayList<String> idRange = new ArrayList<String>();
		try {
			if (Constants.INDEX_LESO.equals(type)) {
				idRange = SQLUtil.getIdRange(connToSrc, Constants.LESO_SRC_TB, "");
			} else if (Constants.INDEX_LESO_EXTRAET.equals(type)) {
				idRange = SQLUtil.getIdRange(connToSrc, Constants.LESO_EXTRAET_SRC_TB, "");
            }else {
				idRange = SQLUtil.getSourceDataIdRange(connToSrc, type);
			}
		} catch (Exception e) {
			log.error("get " + type + " id range error!", e);
			bean.setErrorMsg("get " + type + " id range error! " + e.getMessage());
			return bean;
		} finally {
            if(!SQLUtil.close(conn, connToSrc)) {
                bean.setErrorMsg(bean.getErrorMsg() + " [ close db connections error! ");
            }
		}

		Long minId = Long.valueOf(idRange.get(0)) - 1;
		Long maxId = Long.valueOf(idRange.get(1));

		log.debug("MaxId : " + maxId + ", MinId : " + minId);

		// put data by thread
		ExecutorService threadPool = Executors.newFixedThreadPool(ConfigFactory.getInt("threadPool.basesize", 30));
		List<DownloadWorker> workList = new ArrayList<DownloadWorker>();
		List<Future<DownloadResult>> results = new ArrayList<Future<DownloadResult>>();

		int pageSize = ConfigFactory.getInt("thread.pageSize", 3000);
		Long totalPage = 0l;
		totalPage = (maxId - minId) / pageSize + 1;
		log.debug("totalPage : " + totalPage);

		for (int i = 1; i <= totalPage; i++) {
			log.debug("create thread : from id=" + (minId + pageSize * (i - 1)) + " to id=" + (minId + pageSize * i));
			DownloadWorker worker = null;

			if (Constants.INDEX_LESO.equals(type)) {
				worker = new DownloadWorker(Constants.LESO_TB, "", type, minId + pageSize * (i - 1), minId + pageSize * i);
			} else if (Constants.INDEX_ALBUM.equals(type)) {
				worker = new DownloadWorker(Constants.ALBUM_TB, "", type, minId + pageSize * (i - 1), minId + pageSize * i);
			} else if (Constants.INDEX_STAR.equals(type)) {
				worker = new DownloadWorker(Constants.STAR_TB, "", type, minId + pageSize * (i - 1), minId + pageSize * i);
			} else if (Constants.INDEX_VIDEO.equals(type)) {
				worker = new DownloadWorker(Constants.VIDEO_TB, "", type, minId + pageSize * (i - 1), minId + pageSize * i);
			} else if (Constants.INDEX_LESO_EXTRAET.equals(type)) {
				worker = new DownloadWorker(Constants.LESO_EXTRAET_SRC_TB, "", type, minId + pageSize * (i - 1), minId + pageSize * i);
            }

			workList.add(worker);
		}

		// get result of threads
		int successCnt = 0;
		int failCnt = 0;

		try {
			results = threadPool.invokeAll(workList);

			for (Future<DownloadResult> future : results) {
				try {
					DownloadResult r = future.get();
					if (!future.isCancelled()) {
						successCnt += r.getRecordsCnt();
						log.info("resolve " + r.getRecordsCnt() + " 条数据, result=" + r.isFlag());
					} else {
						failCnt += r.getRecordsCnt();
						log.info("cancel " + r.getRecordsCnt() + " 条数据, result=" + r.isFlag());
					}
				} catch (ExecutionException e) {
					log.error("get thread result ExecutionException!", e);
					bean.setErrorMsg(bean.getErrorMsg() + " [get thread result ExecutionException!"
					        + e.getMessage());
					continue;
				} catch (Exception e) {
					log.error("get thread result exception!", e);
					bean.setErrorMsg(bean.getErrorMsg() + " [ get thread result exception!"
					        + e.getMessage());
					continue;
				}
			}
		} catch (InterruptedException e) {
			log.error("excute " + type + " thread error!", e);
			bean.setErrorMsg(bean.getErrorMsg() + " [excute " + type + " thread error!"
			        + e.getMessage());
		} finally {
			threadPool.shutdown();
			log.info("threadPool " + type + " shutdown!");
		}

		log.info("download " + type + " done!");

		bean.setSubTaskMysqlSuccessedCountNum(successCnt);
		bean.setSubTaskMysqlFailedCountNum(failCnt);

		return bean;
    }

}